package com.subfg.common.rabbitmq.produer;

import com.alibaba.fastjson.JSONObject;
import com.subfg.common.dao.ZlogMQDao;
import com.subfg.common.entity.common.BatchRecordModel;
import com.subfg.common.rabbitmq.constant.QueueContentModel;
import com.subfg.common.rabbitmq.util.RabbitCallbackConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 *  队列发送类
 */
@Component
public class MQSender {
    private Logger log = LoggerFactory.getLogger(MQSender.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private RabbitCallbackConfig rabbitCallbackConfig;
    @Autowired
    private ZlogMQDao zlogMQDao;

    /**
     * 发送消息 无延迟的
     * @param exchange      路由交换机名称
     * @param routingKey    路由关键字,精确的
     * @param message       消息类容
     */
    public void sendMQMessage(String exchange, String routingKey, QueueContentModel message){
        CorrelationData correlationData = new CorrelationData(message.getId());
        rabbitTemplate.setConfirmCallback(rabbitCallbackConfig);
        rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
        log.info("rabbitmq开始发送消息>>>交换机名称==>{}, routingKey==>{}", exchange, routingKey);
        try {
            insertMQMessage(exchange, routingKey, message);
        } catch (Exception e) {
           log.error("写入队列数据失败, 严重错误,可能会导致数据丢失,==>>{}", e);
        }

    }

    /**
     * 向rabbitmq发送延迟队列消息
     * @param exchange       交换机名称
     * @param routingKey     routing key 路由关键字,精确的
     */
    public void sendDelayedMQMessage(String exchange, String routingKey, QueueContentModel sendMessage) {
        CorrelationData correlationData = new CorrelationData(sendMessage.getId());
        rabbitTemplate.convertAndSend(exchange, routingKey, sendMessage, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setHeader("x-delay", sendMessage.getDelayTime());
                return message;
            }
        }, correlationData);
        log.info("rabbitmq开始发送消息>>>交换机名称==>{}, routingKey==>{}, 延迟==>{}毫秒, 发送的消息是==>{}", exchange, routingKey, sendMessage.getDelayTime());
        try {
            insertMQMessage(exchange, routingKey, sendMessage);
        } catch (Exception e) {
            log.error("写入队列数据失败, 严重错误,可能会导致数据丢失", e);
        }
    }
    /** 写入队列 **/
    public void insertMQMessage(String exchange, String routingKey, QueueContentModel sendMessage) {
        if (sendMessage.getCount() == 0) {
            sendMessage.setExchange(exchange);
            sendMessage.setQueueName(routingKey);
            sendMessage.setData(JSONObject.toJSONString(sendMessage.getData()));
            zlogMQDao.insertZlogMQDao(new BatchRecordModel<QueueContentModel>(sendMessage));
        }
    }
}
